package com.atguigu.realtime.util;

import com.atguigu.realtime.common.Constant;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * @auther yuzehua
 * @date 2022/10/24 - 19:20
 */
public class FlinkSourceUtil {
    public static SourceFunction<String> getKafkaSource(String topic, String groupId) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", Constant.KAFKA_BROKERS);
        props.setProperty("group.id", groupId);
        FlinkKafkaConsumer<String> source = new FlinkKafkaConsumer<>(
                topic,
                new KafkaDeserializationSchema<String>() {
                    // 是否要结束流
                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        // 从 kafka 消费数据, 流永远不停, 返回 false
                        return false;
                    }

                    // 对传入的数组, 实现反序列化
                    @Override
                    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                        byte[] value = record.value();
                        // 如果 value 是 null, 不需要反序列化
                        if (value != null) {
                            return new String(value, StandardCharsets.UTF_8);
                        }
                        return null;
                    }

                    // 读到数据的类型, 决定了流中的数据类型
                    @Override
                    public TypeInformation<String> getProducedType() {
//                    return Types.STRING;
//                    return TypeInformation.of(String.class);
                        return TypeInformation.of(new TypeHint<String>() {});
                    }
                },
                props
        );
        source.setStartFromLatest();
        return source;
    }
}
